package com.hucais.etl.job.service

import java.util

import com.hucais.core.constant.Constants
import com.hucais.core.utils.DefaultPropertiesUtil
import com.hucais.etl.common.bean.CandidateCategory
import com.hucais.etl.common.dao.{MysqlDao, OdsHiveDao}
import com.hucais.etl.common.service.CommonQueryService
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark.sql.EsSparkSQL

object DdService {
  /**
   *
   * @param ssc
   * @param sparkSession
   */
  def action(ssc: SparkContext, sparkSession: SparkSession, startDate: String, endDate: String): Unit = {
    // 获取候选分类名单
    val categoryList = MysqlDao.getCandidateCategoryList
    val categoryListBd: Broadcast[util.List[CandidateCategory]] = ssc.broadcast(categoryList)

    // 获取ODS数据
    val bookDS = OdsHiveDao.getOdsBook(sparkSession, Constants.SECOND_CHANNEL_DD, startDate, endDate)
    val storeDS = OdsHiveDao.getOdsStore(sparkSession, Constants.SECOND_CHANNEL_DD, startDate, endDate)
    val commentDS = OdsHiveDao.getOdsComment(sparkSession, Constants.SECOND_CHANNEL_DD, startDate, endDate)

    bookDS.cache()
    storeDS.cache()
    commentDS.cache()
    commentDS.createOrReplaceTempView(Constants.SPARK_VIEW_COMMENT)

    // 清洗数据
    val tmpBookDS = CommonQueryService.defultCleanBookData(sparkSession, bookDS, categoryListBd)
    tmpBookDS.cache()
    tmpBookDS.createOrReplaceTempView(Constants.SPARK_VIEW_BOOK)

    val tmpStroeDS = CommonQueryService.defultCleanStoreData(sparkSession, storeDS)
    tmpStroeDS.cache()
    tmpStroeDS.createOrReplaceTempView(Constants.SPARK_VIEW_STORE)

    // 聚合全量数据
    val tmpBookBaseInfoDS = CommonQueryService.defaultAggAllData(sparkSession)
    tmpBookBaseInfoDS.cache()

    // 组装Result数据输出到ES
    val resultDS = CommonQueryService.initResultData(sparkSession, tmpBookBaseInfoDS, Constants.FIRST_CHANNEL_REVERSE_SELECT, Constants.SECOND_CHANNEL_DD)
    EsSparkSQL.saveToEs(resultDS, DefaultPropertiesUtil.get("dws.book.query.info"))
  }


}
